package com.xyy.saas.payment.util.mq;

import com.xyy.saas.payment.callback.core.dto.CallbackMsg;
import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.common.constants.BaseConstants;
import com.xyy.saas.payment.cores.vo.CallbackResultVo;
import com.xyy.saas.payment.cores.vo.ResultVO;
import com.xyy.saas.payment.util.CallbackUtil;
import com.xyy.saas.payment.util.JSONUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @Author: wuqiongda
 * @Date: 2020/4/20 18:14
 */
@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_CALLBACK_RESULT, consumerGroup = RocketMqGroup.GROUP_CALLBACK_RESULT, consumeMode = ConsumeMode.ORDERLY)
public class CallbackResultConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {


    private static final Logger LOGGER = LoggerFactory.getLogger(CallbackConsumer.class);

    @Autowired
    CallbackUtil callbackUtil;


    @Override
    public void onMessage(MessageExt message) {
        String body = new String(message.getBody());
        try {
            CallbackMsg callbackMsg = JSONUtils.parse2Object(body, CallbackMsg.class);
            LOGGER.info("CallbackResultConsumer#onMessage body:{}",  body);
            CallbackResultVo callbackResultVo = new CallbackResultVo(callbackMsg.getTradeNo(), callbackMsg.getType(), callbackMsg.getAmount(), callbackMsg.getFee(),
                    callbackMsg.getSuccessTime(), callbackMsg.getChannel(), callbackMsg.getBusinessId(), callbackMsg.getSettleStatus(), callbackMsg.getMerchantId());
            // 发送callback request 给 业务
            ResultVO resultVO = ResultVO.createSuccess();
            resultVO.setMsg(callbackMsg.getMsg());
            resultVO.setResult(callbackResultVo);
            callbackUtil.sendResultMsg(callbackMsg.getCallback(), resultVO, BaseConstants.BEGIN_COUNT);
        } catch (Exception e) {
            LOGGER.error("CallbackResultConsumer#onMessage consumer error body:{}",body, e);
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }
}
